use std::{
    collections::{HashSet, hash_map},
    error::Error,
    hash::{Hash, Hasher}, // 引入 std::hash::Hasher 和 std::hash::Hash trait 来实现对 Vec<u8> 的哈希计算
    time::Duration,
};

use libp2p::{
    PeerId, StreamProtocol,
    core::Multiaddr,
    gossipsub::{self, MessageId},
    identify, identity, kad, mdns,
    request_response::{self, ProtocolSupport, ResponseChannel},
    swarm::NetworkBehaviour,
};

use tokio::sync::oneshot;
use tracing::info;

use crate::ProvidingContentId;
use crate::network_core::P2pNetWorkError;
use crate::network_core::network_core_impl::combined_behaviour::{
    ContentRequest, ContentResponse, KadEventMetaData, RequestResponseEventMetaData,
};

// RFC 6762: Multicast DNS - RFC Editor
// RFC 7049: CBOR（Concise Binary Object Representation）
// We create a custom network behaviour that combines Gossipsub, Mdns etc.
#[derive(NetworkBehaviour)]
pub(crate) struct CombinedBehaviour {
    // 使用 libp2p 的 mdns 或 kademlia 协议来发现网络中的其他节点。
    // Peer Discovery with Identify. mdns 和 kad 是两种不同的发现和路由机制
    //  mDNS 主要用于在本地网络（如家庭或办公室网络）中发现其他对等节点。
    //  Kademlia DHT 设计用于在全球范围内查找节点和内容。
    pub(super) kademlia: kad::Behaviour<kad::store::MemoryStore>,
    pub(super) mdns: mdns::tokio::Behaviour,

    // 使用 libp2p 的 request-response 协议来实现区块数据的传输
    pub(super) request_response: request_response::cbor::Behaviour<ContentRequest, ContentResponse>,

    // gossipsub 协议来实现对区块数据的广播和接收, gossipsub 比 flooding  有更好的效率和扩展性
    // Gossipsub is a P2P pubsub (publish/subscription) routing layer designed to extend upon floodsub and meshsub routing protocols.
    pub(super) gossipsub: gossipsub::Behaviour, // for blockchain broadcast

    pub(super) identify: identify::Behaviour,
}

impl CombinedBehaviour {
    pub(crate) fn new(
        key: &identity::Keypair,
    ) -> Result<CombinedBehaviour, Box<dyn Error + Send + Sync>> {
        let local_peer_id = key.public().to_peer_id();

        let kademlia =
            kad::Behaviour::new(local_peer_id, kad::store::MemoryStore::new(local_peer_id));
        let request_response = request_response::cbor::Behaviour::new(
            [(
                StreamProtocol::new("/mini-bc-data-exchange/1"),
                ProtocolSupport::Full,
            )],
            request_response::Config::default(),
        );

        let mdns_config = mdns::Config::default();
        let mdns = mdns::tokio::Behaviour::new(mdns_config, local_peer_id)?;

        // To content-address message, we can take the hash of message and use it as an ID.
        let message_id_fn = |message: &gossipsub::Message| {
            let mut s = hash_map::DefaultHasher::new();
            message.data.hash(&mut s);
            gossipsub::MessageId::from(s.finish().to_string())
        };

        // Set a custom gossipsub configuration
        let gossipsub_config = gossipsub::ConfigBuilder::default()
            .heartbeat_interval(Duration::from_secs(10)) // This is set to aid debugging by not cluttering the log space
            .validation_mode(gossipsub::ValidationMode::Strict) // This sets the kind of message validation.
            .message_id_fn(message_id_fn) // content-address messages. No two messages of the same content will be propagated.
            // .max_transmit_size(2 * 1024 * 1024) // 可选：设置最大传输大小
            .build()
            .map_err(tokio::io::Error::other)?; // Temporary hack because `build` does not return a proper `std::error::Error`.

        // build a gossipsub network behaviour
        let gossipsub = gossipsub::Behaviour::new(
            gossipsub::MessageAuthenticity::Signed(key.clone()),
            gossipsub_config,
        )?;

        let identify = identify::Behaviour::new(identify::Config::new(
            "/mini-bc/0.1.0".to_string(),
            key.public(),
        ));

        Ok(CombinedBehaviour {
            kademlia,
            mdns,
            request_response,
            gossipsub,
            identify,
        })
    }

    pub(crate) fn cleanup(&mut self) {
        // self.kademlia.cleanup();
        // self.mdns.cleanup();
        // self.request_response.cleanup();
        // self.gossipsub.cleanup();
    }
}

// for gossipsub
impl CombinedBehaviour {
    // "mini-bc-boradcast"
    pub(crate) fn gossipsub_subscribe(&mut self, topic: &str) -> Result<bool, P2pNetWorkError> {
        info!("gossipsub subscribe");
        // Create a Gossipsub topic
        let topic = gossipsub::IdentTopic::new(topic);

        // subscribes to our topic
        let subscribed = self.gossipsub.subscribe(&topic)?;
        if subscribed {
            info!("Subscribed to Gossipsub topic: {}", topic);
        } else {
            info!("Already subscribed to Gossipsub topic: {}", topic);
        }

        Ok(subscribed)
    }

    pub(crate) fn gossipsub_publish(
        &mut self,
        topic: &str,
        data: Vec<u8>,
    ) -> Result<MessageId, P2pNetWorkError> {
        info!("gossipsub publish");
        // Create a Gossipsub topic
        let topic_ = gossipsub::IdentTopic::new(topic);

        // publish a message to the network
        let message_id = self.gossipsub.publish(topic_, data)?;
        info!(
            "Successfully published message (ID: {}) to topic {}",
            message_id, topic
        );
        Ok(message_id)
    }
}

// for kademlia
impl CombinedBehaviour {
    pub(crate) fn set_kademlia_mode(&mut self, mode: Option<kad::Mode>) {
        self.kademlia.set_mode(mode);
    }

    pub(crate) fn add_address_kademlia(&mut self, peer_id: &PeerId, addr: Multiaddr) {
        self.kademlia.add_address(peer_id, addr);
    }

    pub(crate) fn start_providing_kademlia(
        &mut self,
        providing_content_id: &ProvidingContentId,
        sender: oneshot::Sender<()>,
        kad_metadata: &mut KadEventMetaData,
    ) {
        info!("start providing kademlia");
        let key = providing_content_id.to_record_key();
        let query_id = self.kademlia.start_providing(key).expect("No store error.");

        kad_metadata
            .pending_start_providing
            .insert(query_id, sender);
    }

    pub(crate) fn get_providers_kademlia(
        &mut self,
        providing_content_id: &ProvidingContentId,
        sender: oneshot::Sender<HashSet<PeerId>>,
        kad_metadata: &mut KadEventMetaData,
    ) {
        info!("get providers kademlia");
        let key = providing_content_id.to_record_key();
        let query_id = self.kademlia.get_providers(key);
        kad_metadata.pending_get_providers.insert(query_id, sender);
    }
}

// for request_response
impl CombinedBehaviour {
    pub(crate) fn send_request_file(
        &mut self,
        providing_content_id: &ProvidingContentId,
        peer: PeerId,
        response_sender: oneshot::Sender<Result<Vec<u8>, P2pNetWorkError>>,
        request_response_metadata: &mut RequestResponseEventMetaData,
    ) {
        info!(
            "send request file({}) to peer {:?}",
            providing_content_id, peer
        );
        let request = ContentRequest(providing_content_id.clone());
        let request_id = self.request_response.send_request(&peer, request);

        request_response_metadata
            .pending_request_file
            .insert(request_id, response_sender);
    }

    pub(crate) fn send_response_file_to_network(
        &mut self,
        content: Result<Vec<u8>, String>,
        channel: ResponseChannel<ContentResponse>,
    ) {
        info!("send response file");
        let response = ContentResponse(content);
        self.request_response
            .send_response(channel, response)
            .expect("Connection to peer to be still open.");
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use libp2p::identity;
    use sha2::{Digest, Sha256};

    use crate::BC_FLOOD_TOPIC_NAME_PREFIX;
    fn crate_user_behaviour() -> (CombinedBehaviour, PeerId) {
        let key = identity::Keypair::generate_ed25519();
        (
            CombinedBehaviour::new(&key).unwrap(),
            key.public().to_peer_id(),
        )
    }

    #[tokio::test]
    async fn test_new_user_behaviour() {
        let key = identity::Keypair::generate_ed25519();
        let behaviour = CombinedBehaviour::new(&key).unwrap();
        assert!(behaviour.kademlia.mode() == kad::Mode::Client);
    }

    #[tokio::test]
    async fn test_gossipsub_subscribe() {
        let (mut behaviour, _peer_id) = crate_user_behaviour();
        let tmp = behaviour.gossipsub_subscribe(BC_FLOOD_TOPIC_NAME_PREFIX);
        assert!(tmp.is_ok());
        assert!(tmp.unwrap());
    }

    #[tokio::test]
    async fn test_set_kademlia_mode() {
        let (mut behaviour, _peer_id) = crate_user_behaviour();
        behaviour.set_kademlia_mode(Some(kad::Mode::Server));
        assert_eq!(behaviour.kademlia.mode(), kad::Mode::Server);
    }

    fn gen_providing_data() -> (String, ProvidingContentId) {
        let data_str = "[request-response-UT] libp2p-request-response!".to_string();
        let hash = Sha256::digest(data_str.as_bytes());
        let hash_bytes: [u8; 32] = hash.into();
        let content_id = ProvidingContentId::new(hash_bytes);

        println!("Gen providing data: `{data_str}`. Hash: {content_id}");
        (data_str, content_id)
    }

    #[tokio::test]
    async fn test_start_providing_kademlia() {
        let (mut behaviour, _peer_id) = crate_user_behaviour();
        let (sender, _) = oneshot::channel();
        let mut kad_metadata = KadEventMetaData::new();
        let (_, providing_content_id) = gen_providing_data();
        behaviour.start_providing_kademlia(&providing_content_id, sender, &mut kad_metadata);
        assert!(!kad_metadata.pending_start_providing.is_empty());
    }

    #[tokio::test]
    async fn test_get_providers_kademlia() {
        let (mut behaviour, _peer_id) = crate_user_behaviour();
        let (sender, _) = oneshot::channel();
        let mut kad_metadata = KadEventMetaData::new();
        let (_, providing_content_id) = gen_providing_data();
        behaviour.get_providers_kademlia(&providing_content_id, sender, &mut kad_metadata);
        assert!(!kad_metadata.pending_get_providers.is_empty());
    }

    #[tokio::test]
    async fn test_handle_mdns_event_discovered() {
        let (mut behaviour, peer_id) = crate_user_behaviour();
        let event = mdns::Event::Discovered(vec![(
            PeerId::random(),
            "/ip4/127.0.0.1/tcp/0".parse().unwrap(),
        )]);
        behaviour.handle_mdns_event(event, &peer_id);
        // This test assumes that the peer has been added, which may require additional checks.
    }

    #[tokio::test]
    async fn test_handle_gossipsub_event_message() {
        let (mut behaviour, peer_id) = crate_user_behaviour();
        let topic = gossipsub::IdentTopic::new("test-net");
        let event = gossipsub::Event::Message {
            propagation_source: PeerId::random(),
            message_id: gossipsub::MessageId::from("test_message_id"),
            message: gossipsub::Message {
                data: "test_message".as_bytes().to_vec(),
                source: None,
                sequence_number: None,
                topic: topic.into(),
            },
        };
        let (event_sender_for_app, _event_receiver_for_app) = tokio::sync::mpsc::channel(2);
        behaviour
            .handle_gossipsub_event(event, &peer_id, &event_sender_for_app)
            .await;
        // This test assumes that the message has been processed, which may require additional checks.
    }
}
